package cn.wolfcode.product._3product_asynchronization;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;

public class ProductTest {

    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException, MQBrokerException {

        //注意生产者生产消息必须在线


        //创建一个生产者对象，并且制定一个生产者组
        DefaultMQProducer producer = new DefaultMQProducer("wolfcode-producer");
        //设置名字服务的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //启动生产者
        producer.start();

        CountDownLatch count = new CountDownLatch(100);

        for (int i = 0; i < 100; i++) {
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("03-producer-type" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " + i).getBytes("utf-8") /* Message body */
            );
            //发送同步消息到一个Broker
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    count.countDown();
                    System.out.println("消息发送成功");
                    System.out.println(JSON.toJSONString(sendResult));
                }

                @Override
                public void onException(Throwable throwable) {
                    count.countDown();
                    System.out.println("消息发送失败"+throwable.getMessage());
                    System.out.println("处理失败消息");

                }
            });

        }
        count.await();
        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();

    }
}
